package drds.binlog.parse.inbound.mysql;

import drds.binlog.LogByteBlock;
import drds.binlog.LogContext;
import drds.binlog.LogDecoder;
import drds.binlog.LogEvent;
import drds.binlog.parse.driver.Connector;
import drds.binlog.parse.driver.QueryExecutor;
import drds.binlog.parse.driver.UpdateExecutor;
import drds.binlog.parse.driver.packets.GtidSet;
import drds.binlog.parse.driver.packets.HeaderPacket;
import drds.binlog.parse.driver.packets.client.BinlogDumpCommandPacket;
import drds.binlog.parse.driver.packets.client.BinlogDumpGtidCommandPacket;
import drds.binlog.parse.driver.packets.client.RegisterSlaveCommandPacket;
import drds.binlog.parse.driver.packets.client.SemiAckCommandPacket;
import drds.binlog.parse.driver.packets.server.ErrorPacket;
import drds.binlog.parse.driver.packets.server.ResultSetPacket;
import drds.binlog.parse.driver.utils.Packets;
import drds.binlog.parse.exception.BinlogParseException;
import drds.binlog.parse.inbound.Connection;
import drds.binlog.parse.inbound.IMultiStageCoprocessor;
import drds.binlog.parse.inbound.SinkFunction;
import drds.binlog.parse.inbound.mysql.dbsync.DirectLogFetcher;
import drds.binlog.parse.support.AuthenticationInfo;
import org.apache.commons.lang.StringUtils;
import org.apache.commons.lang.math.NumberUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.nio.charset.Charset;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;

public class MysqlConnection implements Connection
{

    private static final Logger logger = LoggerFactory.getLogger(MysqlConnection.class);
    protected int connTimeout = 5 * 1000;                                      // 5秒
    protected int soTimeout = 60 * 60 * 1000;                                // 1小时
    private Connector connector;
    private long slaveId;
    private Charset charset = Charset.forName("UTF-8");
    private BinlogFormat binlogFormat;
    private BinlogImage binlogImage;
    // tsdb releated
    private AuthenticationInfo authenticationInfo;
    // dump binlog bytes, 暂不包括meta与TSDB
    private AtomicLong receivedBinlogBytes;

    public MysqlConnection()
    {
    }

    public MysqlConnection(InetSocketAddress inetSocketAddress, String username, String password)
    {
        authenticationInfo = new AuthenticationInfo();
        authenticationInfo.setInetSocketAddress(inetSocketAddress);
        authenticationInfo.setUsername(username);
        authenticationInfo.setPassword(password);
        connector = new Connector(inetSocketAddress, username, password);
        // 将connection里面的参数透传下
        connector.setSoTimeout(soTimeout);
        connector.setConnTimeout(connTimeout);
    }

    public MysqlConnection(InetSocketAddress inetSocketAddress, String username, String password, byte charsetNumber,
                           String defaultSchema)
    {
        authenticationInfo = new AuthenticationInfo();
        authenticationInfo.setInetSocketAddress(inetSocketAddress);
        authenticationInfo.setUsername(username);
        authenticationInfo.setPassword(password);
        authenticationInfo.setDefaultDatabaseName(defaultSchema);
        connector = new Connector(inetSocketAddress, username, password, charsetNumber, defaultSchema);
        // 将connection里面的参数透传下
        connector.setSoTimeout(soTimeout);
        connector.setConnTimeout(connTimeout);
    }

    public void connect() throws IOException
    {
        connector.connect();
    }

    public void reconnect() throws IOException
    {
        connector.reconnect();
    }

    public void disconnect() throws IOException
    {
        connector.disconnect();
    }

    public boolean isConnected()
    {
        return connector.isConnected();
    }

    public ResultSetPacket query(String queryString) throws IOException
    {
        QueryExecutor queryExecutor = new QueryExecutor(connector);
        return queryExecutor.query(queryString);
    }

    public List<ResultSetPacket> querys(String queryString) throws IOException
    {
        QueryExecutor queryExecutor = new QueryExecutor(connector);
        return queryExecutor.querys(queryString);
    }

    public void update(String queryString) throws IOException
    {
        UpdateExecutor updateExecutor = new UpdateExecutor(connector);
        updateExecutor.update(queryString);
    }

    /**
     * 加速主备切换时的查找速度，做一些特殊优化，比如只解析事务头或者尾
     */
    public void seek(String binlogFileName, Long binlogPosition, SinkFunction sinkFunction) throws IOException
    {
        updateSettings();

        sendBinlogDump(binlogFileName, binlogPosition);
        DirectLogFetcher directLogFetcher = new DirectLogFetcher(connector.getReceiveBufferSize());
        directLogFetcher.start(connector.getSocketChannel());
        LogDecoder logDecoder = new LogDecoder();
        logDecoder.handle(LogEvent.ROTATE_EVENT);
        logDecoder.handle(LogEvent.FORMAT_DESCRIPTION_EVENT);
        logDecoder.handle(LogEvent.QUERY_EVENT);
        logDecoder.handle(LogEvent.XID_EVENT);
        LogContext logContext = new LogContext();
        while (directLogFetcher.fetch())
        {
            accumulateReceivedBytes(directLogFetcher.limit());
            LogEvent logEvent = null;
            logEvent = logDecoder.decode(directLogFetcher, logContext);

            if (logEvent == null)
            {
                throw new BinlogParseException("parse failed");
            }

            if (!sinkFunction.sink(logEvent))
            {
                break;
            }
        }
    }

    public void dump(String binlogFileName, Long binlogPosition, SinkFunction sinkFunction) throws IOException
    {
        updateSettings();
        sendRegisterSlave();
        sendBinlogDump(binlogFileName, binlogPosition);
        DirectLogFetcher directLogFetcher = new DirectLogFetcher(connector.getReceiveBufferSize());
        directLogFetcher.start(connector.getSocketChannel());
        LogDecoder logDecoder = new LogDecoder(LogEvent.UNKNOWN_EVENT, LogEvent.ENUM_END_EVENT);
        LogContext logContext = new LogContext();
        while (directLogFetcher.fetch())
        {
            accumulateReceivedBytes(directLogFetcher.limit());
            LogEvent logEvent = null;
            logEvent = logDecoder.decode(directLogFetcher, logContext);

            if (logEvent == null)
            {
                throw new BinlogParseException("parse failed");
            }

            if (!sinkFunction.sink(logEvent))
            {
                break;
            }

            if (logEvent.getSemival() == 1)
            {
                sendSemiAck(logContext.getLogPosition().getFileName(), logContext.getLogPosition().getPosition());
            }
        }
    }

    @Override
    public void dump(GtidSet gtidSet, SinkFunction sinkFunction) throws IOException
    {
        updateSettings();
        sendBinlogDumpGTID(gtidSet);

        DirectLogFetcher directLogFetcher = new DirectLogFetcher(connector.getReceiveBufferSize());
        try
        {
            directLogFetcher.start(connector.getSocketChannel());
            LogDecoder logDecoder = new LogDecoder(LogEvent.UNKNOWN_EVENT, LogEvent.ENUM_END_EVENT);
            LogContext logContext = new LogContext();
            // fix bug: #890 将gtid传输至context中，供decode使用
            logContext.setGtidSet(gtidSet);
            while (directLogFetcher.fetch())
            {
                accumulateReceivedBytes(directLogFetcher.limit());
                LogEvent logEvent = null;
                logEvent = logDecoder.decode(directLogFetcher, logContext);

                if (logEvent == null)
                {
                    throw new BinlogParseException("parse failed");
                }

                if (!sinkFunction.sink(logEvent))
                {
                    break;
                }
            }
        } finally
        {
            directLogFetcher.close();
        }
    }

    public void dump(long timestamp, SinkFunction sinkFunction) throws IOException
    {
        throw new NullPointerException("Not implement yet");
    }

    @Override
    public void dump(String binlogFileName, Long binlogPosition, IMultiStageCoprocessor IMultiStageCoprocessor) throws IOException
    {
        updateSettings();
        sendRegisterSlave();
        sendBinlogDump(binlogFileName, binlogPosition);
        ((MultiStageCoprocessor) IMultiStageCoprocessor).setConnection(this);
        DirectLogFetcher directLogFetcher = new DirectLogFetcher(connector.getReceiveBufferSize());
        try
        {
            directLogFetcher.start(connector.getSocketChannel());
            while (directLogFetcher.fetch())
            {
                accumulateReceivedBytes(directLogFetcher.limit());
                LogByteBlock logByteBlock = directLogFetcher.duplicate();
                directLogFetcher.consume(directLogFetcher.limit());
                if (!IMultiStageCoprocessor.publish(logByteBlock))
                {
                    break;
                }
            }
        } finally
        {
            directLogFetcher.close();
        }
    }

    @Override
    public void dump(long timestamp, IMultiStageCoprocessor multiStageCoprocessor) throws IOException
    {
        throw new NullPointerException("Not implement yet");
    }

    @Override
    public void dump(GtidSet gtidSet, IMultiStageCoprocessor multiStageCoprocessor) throws IOException
    {
        updateSettings();
        sendBinlogDumpGTID(gtidSet);

        ((MultiStageCoprocessor) multiStageCoprocessor).setConnection(this);
        DirectLogFetcher directLogFetcher = new DirectLogFetcher(connector.getReceiveBufferSize());
        try
        {
            directLogFetcher.start(connector.getSocketChannel());
            while (directLogFetcher.fetch())
            {
                accumulateReceivedBytes(directLogFetcher.limit());
                LogByteBlock logByteBlock = directLogFetcher.duplicate();
                directLogFetcher.consume(directLogFetcher.limit());
                if (!multiStageCoprocessor.publish(logByteBlock))
                {
                    break;
                }
            }
        } finally
        {
            directLogFetcher.close();
        }
    }

    private void sendRegisterSlave() throws IOException
    {
        RegisterSlaveCommandPacket registerSlaveCommandPacket = new RegisterSlaveCommandPacket();
        SocketAddress socketAddress = connector.getSocketChannel().getLocalSocketAddress();
        if (socketAddress == null || !(socketAddress instanceof InetSocketAddress))
        {
            return;
        }

        InetSocketAddress inetSocketAddress = (InetSocketAddress) socketAddress;
        String host = inetSocketAddress.getHostString();
        int port = inetSocketAddress.getPort();
        registerSlaveCommandPacket.reportHost = host;
        registerSlaveCommandPacket.reportPort = port;
        registerSlaveCommandPacket.reportPasswd = authenticationInfo.getPassword();
        registerSlaveCommandPacket.reportUser = authenticationInfo.getUsername();
        registerSlaveCommandPacket.serverId = this.slaveId;
        byte[] cmdBody = registerSlaveCommandPacket.toBytes();

        logger.info("Register slave {}", registerSlaveCommandPacket);

        HeaderPacket headerPacket = new HeaderPacket();
        headerPacket.setPacketBodyLength(cmdBody.length);
        headerPacket.setPacketSequenceNumber((byte) 0x00);
        Packets.write(connector.getSocketChannel(), headerPacket.toBytes(), cmdBody);

        headerPacket = Packets.readHeader(connector.getSocketChannel(), 4);
        byte[] body = Packets.readBytes(connector.getSocketChannel(), headerPacket.getPacketBodyLength());
        assert body != null;
        if (body[0] < 0)
        {
            if (body[0] == -1)
            {
                ErrorPacket err = new ErrorPacket();
                err.fromBytes(body);
                throw new IOException("Error When doing Register slave:" + err.toString());
            } else
            {
                throw new IOException("unpexpected packet with field_count=" + body[0]);
            }
        }
    }

    private void sendBinlogDump(String binlogfilename, Long binlogPosition) throws IOException
    {
        BinlogDumpCommandPacket binlogDumpCommandPacket = new BinlogDumpCommandPacket();
        binlogDumpCommandPacket.binlogFileName = binlogfilename;
        binlogDumpCommandPacket.binlogPosition = binlogPosition;
        binlogDumpCommandPacket.slaveServerId = this.slaveId;
        byte[] cmdBody = binlogDumpCommandPacket.toBytes();

        logger.info("COM_BINLOG_DUMP with index$originIndex:{}", binlogDumpCommandPacket);
        HeaderPacket headerPacket = new HeaderPacket();
        headerPacket.setPacketBodyLength(cmdBody.length);
        headerPacket.setPacketSequenceNumber((byte) 0x00);
        Packets.write(connector.getSocketChannel(), headerPacket.toBytes(), cmdBody);
        connector.setDumping(true);
    }

    public void sendSemiAck(String binlogfilename, Long binlogPosition) throws IOException
    {
        SemiAckCommandPacket semiAckCommandPacket = new SemiAckCommandPacket();
        semiAckCommandPacket.binlogFileName = binlogfilename;
        semiAckCommandPacket.binlogPosition = binlogPosition;

        byte[] cmdBody = semiAckCommandPacket.toBytes();

        logger.info("SEMI ack with index$originIndex:{}", semiAckCommandPacket);
        HeaderPacket headerPacket = new HeaderPacket();
        headerPacket.setPacketBodyLength(cmdBody.length);
        headerPacket.setPacketSequenceNumber((byte) 0x00);
        Packets.write(connector.getSocketChannel(), headerPacket.toBytes(), cmdBody);
    }

    private void sendBinlogDumpGTID(GtidSet gtidSet) throws IOException
    {
        BinlogDumpGtidCommandPacket binlogDumpGtidCommandPacket = new BinlogDumpGtidCommandPacket();
        binlogDumpGtidCommandPacket.slaveServerId = this.slaveId;
        binlogDumpGtidCommandPacket.gtidSet = gtidSet;
        byte[] cmdBody = binlogDumpGtidCommandPacket.toBytes();

        logger.info("COM_BINLOG_DUMP_GTID:{}", binlogDumpGtidCommandPacket);
        HeaderPacket headerPacket = new HeaderPacket();
        headerPacket.setPacketBodyLength(cmdBody.length);
        headerPacket.setPacketSequenceNumber((byte) 0x00);
        Packets.write(connector.getSocketChannel(), headerPacket.toBytes(), cmdBody);
        connector.setDumping(true);
    }

    public MysqlConnection fork()
    {
        MysqlConnection mysqlConnection = new MysqlConnection();
        mysqlConnection.setCharset(getCharset());
        mysqlConnection.setSlaveId(getSlaveId());
        mysqlConnection.setConnector(connector.fork());
        // set authenticationInfo
        mysqlConnection.setAuthenticationInfo(authenticationInfo);
        return mysqlConnection;
    }

    @Override
    public long queryServerId() throws IOException
    {
        ResultSetPacket resultSetPacket = query("show variables like 'server_id'");
        List<String> fieldValues = resultSetPacket.getFieldValueList();
        if (fieldValues == null || fieldValues.size() != 2)
        {
            return 0;
        }
        return NumberUtils.toLong(fieldValues.get(1));
    }

    // ====================== help method ====================

    /**
     * the settings that will need to be checked or set:<br>
     * <ol>
     * <li>wait_timeout</li>
     * <li>net_write_timeout</li>
     * <li>net_read_timeout</li>
     * </ol>
     *
     * @throws IOException
     */
    private void updateSettings() throws IOException
    {
        try
        {
            update("set wait_timeout=9999999");
        } catch (Exception e)
        {
            logger.warn("update wait_timeout failed", e);
        }
        try
        {
            update("set net_write_timeout=1800");
        } catch (Exception e)
        {
            logger.warn("update net_write_timeout failed", e);
        }

        try
        {
            update("set net_read_timeout=1800");
        } catch (Exception e)
        {
            logger.warn("update net_read_timeout failed", e);
        }

        try
        {
            // 设置服务端返回结果时不做编码转化，直接按照数据库的二进制编码进行发送，由客户端自己根据需求进行编码转化
            update("set names 'binary'");
        } catch (Exception e)
        {
            logger.warn("update names failed", e);
        }

        try
        {
            // mysql5.6针对checksum支持需要设置session变量
            // 如果不设置会出现错误： Slave can not handle replication events with the
            // checksum that master is configured to log
            // 但也不能乱设置，需要和mysql server的checksum配置一致，不然RotateLogEvent会出现乱码
            update("set @master_binlog_checksum= '@@global.binlog_checksum'");
        } catch (Exception e)
        {
            logger.warn("update master_binlog_checksum failed", e);
        }

        try
        {
            // 参考:https://github.com/alibaba/canal/issues/284
            // mysql5.6需要设置slave_uuid避免被server kill链接
            update("set @slave_uuid=uuid()");
        } catch (Exception e)
        {
            if (!StringUtils.contains(e.getMessage(), "Unknown system variable"))
            {
                logger.warn("update slave_uuid failed", e);
            }
        }

        try
        {
            // mariadb针对特殊的类型，需要设置session变量
            update("SET @mariadb_slave_capability='" + LogEvent.MARIA_SLAVE_CAPABILITY_MINE + "'");
        } catch (Exception e)
        {
            logger.warn("update mariadb_slave_capability failed", e);
        }

        /**
         * MASTER_HEARTBEAT_PERIOD sets the interval in seconds between
         * replication heartbeats. Whenever the master's binary log is updated
         * with an event, the waiting period for the next heartbeat is reset.
         * interval is a decimal value having the range 0 to 4294967 seconds and
         * a resolution in milliseconds; the smallest nonzero value is 0.001.
         * Heartbeats are sent by the master only if there are no unsent events
         * in the binary log file for a period longer than interval.
         */
        try
        {
            long periodNano = TimeUnit.SECONDS.toNanos(DirectLogFetcher.master_heartbeat_period_seconds);
            update("SET @master_heartbeat_period=" + periodNano);
        } catch (Exception e)
        {
            logger.warn("update master_heartbeat_period failed", e);
        }
    }

    /**
     * 获取一下binlog format格式
     */
    private void loadBinlogFormat()
    {
        ResultSetPacket resultSetPacket = null;
        try
        {
            resultSetPacket = query("show variables like 'binlog_format'");
        } catch (IOException e)
        {
            throw new BinlogParseException(e);
        }

        List<String> columnValues = resultSetPacket.getFieldValueList();
        if (columnValues == null || columnValues.size() != 2)
        {
            logger.warn("unexpected binlog format query result, this may cause unexpected result, so throw binlogParseException to request network to io shutdown.");
            throw new IllegalStateException("unexpected binlog format query result:" + resultSetPacket.getFieldValueList());
        }

        binlogFormat = BinlogFormat.valuesOf(columnValues.get(1));
        if (binlogFormat == null)
        {
            throw new IllegalStateException("unexpected binlog format query result:" + resultSetPacket.getFieldValueList());
        }
    }

    /**
     * 获取一下binlog image格式
     */
    private void loadBinlogImage()
    {
        ResultSetPacket resultSetPacket = null;
        try
        {
            resultSetPacket = query("show variables like 'binlog_row_image'");
        } catch (IOException e)
        {
            throw new BinlogParseException(e);
        }

        List<String> columnValues = resultSetPacket.getFieldValueList();
        if (columnValues == null || columnValues.size() != 2)
        {
            // 可能历时版本没有image特性
            binlogImage = BinlogImage.FULL;
        } else
        {
            binlogImage = BinlogImage.valuesOf(columnValues.get(1));
        }

        if (binlogFormat == null)
        {
            throw new IllegalStateException("unexpected binlog image query result:" + resultSetPacket.getFieldValueList());
        }
    }

    private void accumulateReceivedBytes(long x)
    {
        if (receivedBinlogBytes != null)
        {
            receivedBinlogBytes.addAndGet(x);
        }
    }

    public Charset getCharset()
    {
        return charset;
    }

    public void setCharset(Charset charset)
    {
        this.charset = charset;
    }

    // ================== setter / getter ===================

    public long getSlaveId()
    {
        return slaveId;
    }

    public void setSlaveId(long slaveId)
    {
        this.slaveId = slaveId;
    }

    public Connector getConnector()
    {
        return connector;
    }

    public void setConnector(Connector connector)
    {
        this.connector = connector;
    }

    public BinlogFormat getBinlogFormat()
    {
        if (binlogFormat == null)
        {
            synchronized (this)
            {
                loadBinlogFormat();
            }
        }

        return binlogFormat;
    }

    public BinlogImage getBinlogImage()
    {
        if (binlogImage == null)
        {
            synchronized (this)
            {
                loadBinlogImage();
            }
        }

        return binlogImage;
    }

    public InetSocketAddress getAddress()
    {
        return authenticationInfo.getInetSocketAddress();
    }

    public void setConnTimeout(int connTimeout)
    {
        this.connTimeout = connTimeout;
    }

    public void setSoTimeout(int soTimeout)
    {
        this.soTimeout = soTimeout;
    }

    public AuthenticationInfo getAuthenticationInfo()
    {
        return authenticationInfo;
    }

    public void setAuthenticationInfo(AuthenticationInfo authenticationInfo)
    {
        this.authenticationInfo = authenticationInfo;
    }

    public void setReceivedBinlogBytes(AtomicLong receivedBinlogBytes)
    {
        this.receivedBinlogBytes = receivedBinlogBytes;
    }

}
